import traceback
from datetime import datetime

import pandas as pd

from air_web.config.config import config
from air_web.web_flask.bll.base_bll import BaseBll
from air_web.web_flask.bll.model_map import ModelMap
from air_web.web_flask.common.constants import (
    ORG_MAP,
    ReturnType,
    low_type_map,
    low_type_map_for_sql,
    low_type_map_for_sql_new,
    low_type_map_new,
)


class CommonBll(BaseBll):
    C_CONS = config["C_CONS"]

    def __init__(self):
        """
        如果当前页面用到mysql则init_mysql_dal=True，self.sql_dal就会提供mysql实例，不用到则传False
        如果当前页面用到es则init_es_dal=True，self.es_dal就会提供es实例，不用到则传False
        """
        super().__init__(init_mysql_dal=True, init_es_dal=True)
        self.today = datetime.now().strftime("%Y-%m-%d")
        self.year = str(datetime.now().year)

    def get_org(
        self,
        page_num=None,
        page_size=None,
        org_name=None,
        org_no=None,
        *args,
        **kwargs,
    ):
        try:
            limit_sql = ""
            if None not in [page_num, page_size]:
                begin_point = (page_num - 1) * page_size
                limit_sql = f" limit {page_size} offset {begin_point} "

            where_sql = ""
            where_sql_list = []
            value = []
            org_no_p_org_name = org_no_real_name = ""
            if org_name is not None:
                where_sql_list.append(" org_name like %s ")
                value.append(f"%{org_name}%")

            level_sql = """select org_no, org_name, p_org_no, org_level
            from real_org_no where org_level=0
            """
            level_result = self.sql_dal.sql_pandas.query(level_sql, params=[])
            level_result = level_result.to_dict("record")
            level_result = level_result[0]
            if org_no is None:
                org_no = level_result.get("org_no")
                where_sql_list.append(" org_level = %s")
                value.append(1)
            if org_no is not None:
                if org_no == level_result.get("org_no"):
                    where_sql_list.append(" org_level = %s")
                    value.append(1)
                else:
                    get_p_org_sql = f"""
                    select a.*, b.p_org_name from (select org_no,  p_org_no, org_name
                    from real_org_no where org_no = '{org_no}' and org_level<3) a left join 
                    (select org_no, org_name as p_org_name from real_org_no ) b on a.p_org_no = b.org_no limit 1
            """
                    p_org_df = self.sql_dal.sql_pandas.query(get_p_org_sql)
                    if len(p_org_df):
                        org_no_p_org_name = p_org_df.to_dict("records")[0][
                            "p_org_name"
                        ]
                        org_no_real_name = p_org_df.to_dict("records")[0][
                            "org_name"
                        ]
                    where_sql_list.append(" org_no like %s")
                    value.append(f"{org_no}__")

            if len(where_sql_list) >= 1:
                where_sql = "and" + " and ".join(where_sql_list)
            old_sql = (
                """select org_no, org_name, p_org_no, org_level 
            from real_org_no where org_level < 3 """
                + where_sql
                + " order by org_no "
                + limit_sql
            )

            sql = f""" select a.*, b.p_org_name from ({old_sql}) a left join 
            (select org_no, org_name as p_org_name from real_org_no ) b on a.p_org_no = b.org_no"""
            df_result = self.sql_dal.sql_pandas.query(sql, params=value)
            result = df_result.to_dict("record")
            org_level = 0
            if org_no is not None:
                org_level_sql = f"""select org_level from real_org_no where org_no = '{org_no}' """
                org_level_df = self.sql_dal.sql_pandas.query(org_level_sql)
                if len(org_level_df) >= 1:
                    org_level = org_level_df.to_dict("records")[0].get(
                        "org_level"
                    )
            return (
                [org_no_real_name, org_no_p_org_name],
                result,
                level_result,
                org_level,
            )
        except:
            self.logger.error(traceback.format_exc())
            return ReturnType.ERROR

    def get_tree(self, relation_list):
        tree_dict, temp_dict = {}, {}
        try:
            for i in relation_list:
                temp_dict[i["type_id"]] = i

            for k, v in temp_dict.items():
                if temp_dict.get(k).get("p_type_id"):
                    if not temp_dict.get(
                        temp_dict.get(k).get("p_type_id")
                    ).get("children"):
                        temp_dict[temp_dict[k].get("p_type_id")][
                            "children"
                        ] = []
                    temp_dict[temp_dict[k].get("p_type_id")][
                        "children"
                    ].append(temp_dict.get(k))
                    # temp_dict[temp_dict[k].get("p_type_id")]["children"].sort(
                    #     key=lambda x: x["type_id"], reverse=False
                    # )
                else:
                    tree_dict[temp_dict.get(k).get("type_id")] = temp_dict.get(
                        k
                    )
        except Exception as e:
            msg = e
        finally:
            return tree_dict

    def get_type_code(self, type_code_sort=None, org_no=None):
        try:
            where_sql = ""
            where_sql_list = []
            value = []
            if type_code_sort is not None:
                where_sql_list.append(" left(type_id,1)=  %s ")
                value.append(type_code_sort)
            if org_no is not None:
                where_sql_list.append(" org_no = %s ")
                value.append(org_no)

            if len(where_sql_list) >= 1:
                where_sql = " where " + " and ".join(where_sql_list)

            if org_no[:2] == '13':  # 河北特殊处理
                sql = f"""select type_id, type_code, p_type_id,type_level
                          from area_type_map atm
                          left join 
                          (
                           select trade_id, sort_no 
                           from trade_code_id_rela where trade_level in (4,5)
                          )tc on atm.type_id=tc.trade_id
                          {where_sql}
                          order by sort_no"""
                df_result = self.sql_dal.sql_pandas.query(sql, params=value)
                result = df_result.to_dict("record")
                result = self.get_tree(result)
                if type_code_sort:
                    result = result.get(type_code_sort).get("children", [])
                else:
                    result = [result.get(1, {}), result.get(2, {})]

            else:
                sql = f"""select type_id, type_code, p_type_id,type_level
                          from area_type_map atm
                          right join 
                          (
                           select trade_id, sort_no 
                           from trade_code_id_rela where trade_level in (4,5)
                          )tc on atm.type_id=tc.trade_id
                          {where_sql}
                          order by sort_no"""

                df_result = self.sql_dal.sql_pandas.query(sql, params=value)
                df_result.loc[df_result["type_level"] == 1, "p_type_id"] = 0
                result = df_result.to_dict("record")
                result = self.get_tree(result)
                result = [v for k, v in result.items()]
            return result
        except:
            self.logger.error(traceback.format_exc())
            return ReturnType.ERROR

    def get_industry_info(self):
        try:
            sql = """select type_id, type_code, type_level 
                      from type_map tm
                      right join 
                      (
                       select trade_id, sort_no 
                       from trade_code_id_rela where trade_level=4
                      )tc on tm.type_id=tc.trade_id
                      where type_level=1 
                      order by sort_no
                  """
            type_df = self.sql_dal.sql_pandas.query(sql)

            industry_map = config["industry_map"]
            type_df["p_type_id"] = type_df["type_id"].replace(
                industry_map["type_id"], industry_map["industry_id"]
            )

            industry_info = [
                [31, "第一产业", 0, 0],
                [32, "第二产业", 0, 0],
                [33, "第三产业", 0, 0],
            ]
            industry_df = pd.DataFrame(
                industry_info,
                columns=["type_id", "type_code", "type_level", "p_type_id"],
            )
            industry_df = pd.concat([industry_df, type_df])
            result = industry_df.to_dict("record")
            result = self.get_tree(result)
            result = (
                [result.get(31, {})]
                + [result.get(32, {})]
                + [result.get(33, {})]
            )
            return result
        except:
            self.logger.error(traceback.format_exc())
            return ReturnType.ERROR

    def get_cons_info(
        self,
        org_no=None,
        org_level=None,
        type_id=None,
        type_level=None,
        cons_name=None,
        type_code_sort=None,
        is_monitor=None,
    ):
        try:
            where_sql = ""
            where_sql_list = []
            value = []
            if org_level:
                where_sql_list.append(
                    " {} = {} ".format(ORG_MAP[org_level], org_no)
                )
                # value.append(org_no)
            if type_id:
                # type_code = ModelMap.get('type_code', type_code, reverse=True)
                if type_level == 1:
                    where_sql_list.append(f" pare_type_id = '{type_id}' ")
                else:
                    where_sql_list.append(f" type_id = '{type_id}' ")
            if cons_name:
                where_sql_list.append(
                    f" (cons_name like '%%{cons_name}%%' or cons_no like '%%{cons_name}%%') "
                )
            if type_code_sort:
                where_sql_list.append(f" type_code_sort = '{type_code_sort}' ")
            # if is_monitor:
            #     where_sql_list.append(f" is_monitor = 1 ")
            if len(where_sql_list) >= 1:
                where_sql = " and " + " and ".join(where_sql_list)
            sql = """select cons_no, 
                            cons_no as real_cons_no,
                            cons_name
                     from {c_cons} where is_remove = 0
                     {where_sql}
                     order by data_rank, cons_name asc limit 100
                  """.format(
                c_cons=self.C_CONS, cons_name=cons_name, where_sql=where_sql
            )
            res_df = self.sql_dal.sql_pandas.query(sql)
            # res_df.drop_duplicates(subset=['cons_name'], inplace=True)
            print(f"get_cons_info :{len(res_df)}")
            # print(res_df.to_dict('records'))
            res_dict = res_df.to_dict("records")
            return res_dict
        except:
            self.logger.error(traceback.format_exc())
            return ReturnType.ERROR

    def get_low_type(self):
        # return {"城市台区": ['城市居民', '城市混合', '城市工商业'],
        #         "农村台区": ['农村居民', '农村混合', '农村灌溉']}
        where_sql = "where type_code_sort = 2 "
        sql = (
            """select distinct type_id, type_code, min(type_id) from area_type_map """
            + where_sql
            + "group by type_id, type_code order by min(data_rank)"
        )
        df_result = self.sql_dal.sql_pandas.query(sql)
        # type_map = ModelMap.get_all('type_code', reverse=True)
        # df_result['type_code'] = df_result['type_id'].replace(type_map)
        result = df_result.to_dict("record")
        return result

    def get_finish_time(self):
        try:
            sql = "select finish_time from dws_finish_time where module_name='dws'"
            df = self.sql_dal.sql_pandas.query(sql)
            if df.empty:
                return ReturnType.EMPTY
            else:
                finish_time = df['finish_time'].dt.strftime('%Y-%m-%d %H:%M:%S').tolist()[0]
                return {"finish_time": finish_time}
        except:
            self.logger.error(traceback.format_exc())
            return ReturnType.ERROR
